package com.atguigu.flink.chapter07_state;

import com.atguigu.flink.pojo.MyUtil;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.contrib.streaming.state.EmbeddedRocksDBStateBackend;
import org.apache.flink.runtime.state.FunctionInitializationContext;
import org.apache.flink.runtime.state.FunctionSnapshotContext;
import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;

import java.util.ArrayList;
import java.util.List;

/**
 * Created by Smexy on 2022/10/26
 *
 *   设置状态后端和CK的API
 */
public class Demo2_StateBackEndApi
{
    public static void main(String[] args) {


        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 3333);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);

        env.setParallelism(2);

        //开启checkpoint,每间隔2s，持久化一次   无限failvoer的restart
        env.enableCheckpointing(4000);

        //状态后端声明  默认
        env.setStateBackend(new HashMapStateBackend());
        env.setStateBackend(new EmbeddedRocksDBStateBackend());

        //设置持久化的位置 默认如果不设置外部的ck目录，把快照ck到JobManager进程的内存中
        //env.getCheckpointConfig().setCheckpointStorage("file:///d:/ck");
       env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:9820/ck");

        //设置cancel后依旧保留快照
        // 外部(把快照存储在外部系统，比如hdfs，本地文件系统)ck是否清理。 cancel后，外部的ck也不会删除
        env.getCheckpointConfig().setExternalizedCheckpointCleanup(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);


        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);

        // 确认 checkpoints 之间的时间会进行 500 ms,barior发送频率
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(500);

        // Checkpoint 必须在一分钟内完成，否则就会被抛弃
        env.getCheckpointConfig().setCheckpointTimeout(60000);

        // 同一时间只允许一个 checkpoint 进行
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);


        env
           .socketTextStream("hadoop103", 8888)
           .map(new MyMapFunction()).name("acc")
           .addSink(new SinkFunction<String>()
           {
               @Override
               public void invoke(String value, Context context) throws Exception {
                   if (value.contains("x")){
                       throw new RuntimeException("出异常了!");
                   }
                   System.out.println(value);
               }
           });

        try {
                    env.execute();
                } catch (Exception e) {
                    e.printStackTrace();
                }
    }

    public static class MyMapFunction implements MapFunction<String,String>, CheckpointedFunction
    {

        private List<String> accList = new ArrayList<>();
        //状态
        private ListState<String> acc;

        @Override
        public String map(String value) throws Exception {
            accList.add(value);
            //将accList放入状态
            //acc.update(accList);  //全量
            acc.add(value);   //增量
            return accList.toString();
        }

        //快照(存档)状态  根据设置的间隔，定期执行
        @Override
        public void snapshotState(FunctionSnapshotContext context) throws Exception {

            System.out.println("MyMapFunction.snapshotState");

            //把当前List<String> accList中的数据存入 ListState<String> acc
            //用的是 ListState<String> acc，也称为ManagedState，只需要把你存档的数据放入 ListState<String> acc，flink自动持久化到指定的路径
            //acc.clear(); //清空当前状态中的所有数据
            //acc.addAll(accList); //放入一个list

            //acc.update(accList);  //覆盖写
        }

        //初始化(Task，一启动就做的事情)状态。 只运行一次。 启动后读取之前存档的状态

        // context.getOperatorStateStore(): 容器，容器里有很多状态。
        // context.getOperatorStateStore().getListState()： 取一个ListState
        @Override
        public void initializeState(FunctionInitializationContext context) throws Exception {
            System.out.println(Thread.currentThread().getName()+"MyMapFunction.initializeState");
           // acc = context.getOperatorStateStore().getListState(new ListStateDescriptor<String>("acc", String.class));
            acc = context.getOperatorStateStore().getUnionListState(new ListStateDescriptor<String>("acc", String.class));

            //状态中存储了一些信息，要把这个信息读出来
             accList = MyUtil.toList(acc.get());
        }
    }
}
